package com.huitone.smspfm.interfacz.mt;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bin.common.utils.DateTimeUtils;
import com.bin.common.utils.MD5Util;
import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.interfacz.core.AppStaticConfigs;
import com.huitone.smspfm.interfacz.core.filewriter.FileRoute;
import com.huitone.smspfm.interfacz.dao.SmsSendRecordMapper;
import com.huitone.smspfm.interfacz.dao.entity.SmsSendRecord;
import com.huitone.smspfm.interfacz.dao.entity.SmsSendRecordExample;
import com.huitone.smspfm.interfacz.logger.ExtLogger;
import com.huitone.smspfm.context.utils.ContextPropertiesUtils;
import com.huitone.smspfm.context.utils.ContextUtils;
import com.huitone.smspfm.interfacz.model.MessageSbt;
import com.huitone.smspfm.interfacz.model.MtOldSms;
import com.huitone.smspfm.interfacz.model.RedUserRecord;
import com.huitone.smspfm.interfacz.model.SmsSpeedStatistics;
import com.huitone.smspfm.interfacz.tasks.FileWriterTask;
import com.huitone.smspfm.redis.model.AppCanstant;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;
import com.huitone.smspfm.socket.core.handler.DefaultMessageHandler;
import com.huitone.smspfm.socket.core.net.ExtSocket;
import com.huitone.smspfm.socket.protocol.sgip.constant.ErrorCode;
import com.huitone.smspfm.socket.protocol.sgip.message.Message;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.ActiveTest;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Submit;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Submit2;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.UnBind;
import com.huitone.smspfm.socket.protocol.sgip.message.msghead.Head;

/**
 * @Title SocketServerMessageHandler.java
 * @Package com.huitone.smspfm.interfacz.core
 * @Description TODO(用一句话描述该文件做什么)
 * @author ozh
 * @date 2018年8月21日 下午3:18:53
 * @version V1.1
 */
public class MtMessageHandler extends DefaultMessageHandler {

	private static final ExtLogger logger = ExtLogger
			.create(MtMessageHandler.class);

	private String iSMCache = ContextPropertiesUtils.getProperty("SMCache");
	private byte[] iSMCacheB = AppStaticConfigs.stringRedisSerializer
			.serialize(iSMCache);

	private String iSMCacheHashPrefix = ContextPropertiesUtils
			.getProperty("SMCacheHashPrefix");

	private String smspfmModuleName = ContextPropertiesUtils
			.getProperty("smspfm.module.name");

	private static final String CurrAcceptSpeedS = "CurrAcceptSpeedS";
	private static final String CurrAcceptSpeedM = "CurrAcceptSpeedM";

	private static final byte[] CURR_ACCEPT_SPEEDS_B = AppStaticConfigs.stringRedisSerializer
			.serialize(CurrAcceptSpeedS);

	private static final byte[] CURR_ACCEPT_SPEEDM_B = AppStaticConfigs.stringRedisSerializer
			.serialize(CurrAcceptSpeedM);

	private byte[] currAcceptSpeedS_List = AppStaticConfigs.stringRedisSerializer
			.serialize(AppStaticConfigs.RunningParamsHash + ":"
					+ CurrAcceptSpeedS);
	private byte[] currAcceptSpeedM_List = AppStaticConfigs.stringRedisSerializer
			.serialize(AppStaticConfigs.RunningParamsHash + ":"
					+ CurrAcceptSpeedM);

	private long currSpeedS = 0;

	private long currSpeedM = 0;

	private final Object slock = new Object();

	private final Object mlock = new Object();

	private final Object speedLock = new Object();

	private SocketServerArrayBlockingQueue sbt1Msg = new SocketServerArrayBlockingQueue(
			1000);
	private SocketServerArrayBlockingQueue sbt2Msg = new SocketServerArrayBlockingQueue(
			1000);

	private ConcurrentLinkedQueue<String> redUserMsg = new ConcurrentLinkedQueue<String>();

	private long redUserMaxTime = 20 * 1000L;// TODO 默认３分钟

	private long redUserSleepTime = 1000L;

	private int redUserSize = 100;

	// private long currTableTime = System.currentTimeMillis();

	private long redUserCurrTime = System.currentTimeMillis();

	private RedUserTask redUserTread = null;

	private ExecutorService executoRedUserrService = new ThreadPoolExecutor(2,
			2, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(1000),
			new RejectedExecutionHandler() {

				@Override
				public void rejectedExecution(Runnable r,
						ThreadPoolExecutor executor) {
					boolean offer = false;
					while (!executor.isShutdown() && !offer) {
						offer = executor.getQueue().offer(r);
						if (!offer) {
							try {
								TimeUnit.MILLISECONDS.sleep(10);
							} catch (InterruptedException e) {
							}
						}
					}
				}
			});

	private ArrayBlockingQueue<Map<byte[], byte[]>> speeds = new ArrayBlockingQueue<Map<byte[], byte[]>>(
			1000);

	private SeqRepeatFilterQueue filterQueue = new SeqRepeatFilterQueue();

	private ExecutorService executorService = new ThreadPoolExecutor(
			AppStaticConfigs.smspfmMhePoolSize,
			AppStaticConfigs.smspfmMhePoolSize, 1L, TimeUnit.MINUTES,
			new ArrayBlockingQueue<>(1000), new RejectedExecutionHandler() {

				@Override
				public void rejectedExecution(Runnable r,
						ThreadPoolExecutor executor) {
					boolean offer = false;
					while (!executor.isShutdown() && !offer) {
						offer = executor.getQueue().offer(r);
						if (!offer) {
							try {
								TimeUnit.MILLISECONDS.sleep(10);
							} catch (InterruptedException e) {
							}
						}
					}
				}
			});

	private ExecutorService executorSaveService = new ThreadPoolExecutor(10,
			10, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(100),
			new RejectedExecutionHandler() {

				@Override
				public void rejectedExecution(Runnable r,
						ThreadPoolExecutor executor) {
					boolean offer = false;
					while (!executor.isShutdown() && !offer) {
						offer = executor.getQueue().offer(r);
						if (!offer) {
							try {
								TimeUnit.MILLISECONDS.sleep(10);
							} catch (InterruptedException e) {
							}
						}
					}
				}
			});

	public MtMessageHandler() {

		if (AppStaticConfigs.speedStatisticsOn) {

			startUpdateSpeedTasks();

			startSpeedWriteTasks();

		}

		saveSbt1();

		saveSbt2();

		startRedUserTask();
	}

	private void startRedUserTask() {
		if (redUserTread == null || !redUserTread.isAlive()) {
			redUserTread = new RedUserTask();
			redUserTread.start();
		}

	}

	private void saveSbt2() {

		new Thread(new Runnable() {

			@Override
			public void run() {

				while ((executorService != null && !executorService
						.isShutdown()) || sbt2Msg.getSize() > 0) {

					List<MessageSbt> list = sbt2Msg.poll();

					if (list != null && !list.isEmpty()) {

						executorSaveService.submit(new Runnable() {

							@Override
							public void run() {

								List<String> serialnos = new ArrayList<>();

								AppStaticConfigs.pipelineOpts
										.save(new ABExcuteCommands() {

											@Override
											public void excute(
													RedisConnection connection) {

												List<byte[]> seqNums = new ArrayList<byte[]>();

												for (int i = 0; i < list.size(); i++) {

													MessageSbt msg = list
															.get(i);

													Head head = (Head) msg
															.getHead();
													Submit2 body = (Submit2) msg
															.getBody();

													String recvTime = DateTimeUtils.dateToStr(
															msg.getRecvTime(),
															"yyyyMMddHHmmss");
													String sendPort = StringUtils.substring(
															body.getSendPort(),
															5); // 剔除端口号(固定前五位),保留扩展码
													// body.getSendPort();

													Map<byte[], byte[]> map = new HashMap<>();

													map.put(AppCanstant.SmsEntity.Srv_node,
															AppStaticConfigs.stringRedisSerializer
																	.serialize(body
																			.getSrvNode()));
													map.put(AppCanstant.SmsEntity.ServiceCode,
															AppStaticConfigs.stringRedisSerializer
																	.serialize(body
																			.getServiceCode()));
													map.put(AppCanstant.SmsEntity.SmType,
															AppStaticConfigs.stringRedisSerializer
																	.serialize("1"));
													map.put(AppCanstant.SmsEntity.User,
															AppStaticConfigs.stringRedisSerializer
																	.serialize(body
																			.getUserNumber()));
													map.put(AppCanstant.SmsEntity.Smcont,
															AppStaticConfigs.stringRedisSerializer
																	.serialize(body
																			.getMsgContent()));
													map.put(AppCanstant.SmsEntity.Reportflag,
															AppStaticConfigs.stringRedisSerializer
																	.serialize(StringUtils
																			.toString(body
																					.getReportFlag())));
													map.put(AppCanstant.SmsEntity.recvTime,
															AppStaticConfigs.stringRedisSerializer
																	.serialize(recvTime));
													map.put(AppCanstant.SmsEntity.RecvModuleName,
															AppStaticConfigs.stringRedisSerializer
																	.serialize(smspfmModuleName));
													map.put(AppCanstant.SmsEntity.SendPort,
															AppStaticConfigs.stringRedisSerializer
																	.serialize(sendPort));
													map.put(AppCanstant.SmsEntity.SerialSegment,
															AppStaticConfigs.stringRedisSerializer.serialize(head
																	.getSerialNoPart1()
																	+ ","
																	+ head.getSerialNoPart2()
																	+ ","
																	+ head.getSerialNoPart3()));
													map.put(AppCanstant.SmsEntity.OrgSerialNo,
															AppStaticConfigs.stringRedisSerializer
																	.serialize(body
																			.getOrgSerialNo()));
													map.put(AppCanstant.SmsEntity.OrgSrvNode,
															AppStaticConfigs.stringRedisSerializer
																	.serialize(body
																			.getOrgSrvNode()));
													map.put(AppCanstant.SmsEntity.SendFeedbackResult,
															AppStaticConfigs.stringRedisSerializer
																	.serialize("0"));

													String serialNo = head
															.getSerialNo();
													String userNumber = body
															.getUserNumber();

													String sequn = serialNo
															+ ":" + userNumber;

													connection
															.hMSet(AppStaticConfigs.stringRedisSerializer
																	.serialize(iSMCacheHashPrefix
																			+ ":"
																			+ sequn),
																	map);
													// connection.lPush(iSMCacheB,
													// AppStaticConfigs.stringRedisSerializer.serialize(sequn));
													// connection.zAdd(AppStaticConfigs.iRptRecvSerialsZSetB,
													// System.currentTimeMillis(),
													// AppStaticConfigs.stringRedisSerializer.serialize(msg.getEncodeSerialNo()));
													seqNums.add(AppStaticConfigs.stringRedisSerializer
															.serialize(sequn));

													serialnos.add(serialNo);
												}

												connection
														.lPush(iSMCacheB,
																seqNums.toArray(new byte[1][]));
											}

											@Override
											public void excuteCallback(
													List<Object> results) {
												logger.acceptd(serialnos.size()
														+ " saved "
														+ serialnos.toString());
											}

											@Override
											public void excuteFailCallback() {
												for (int i = 0; i < list.size(); i++) {
													MessageSbt msg = list.get(i);
													Head head = (Head) msg
															.getHead();
													Submit2 body = (Submit2) msg
															.getBody();
													String serialNo = head.getSerialNo();
													if(serialnos.contains(serialNo)){
														continue;
													}
													String recvTime = DateTimeUtils.dateToStr(
															msg.getRecvTime(),
															"yyyyMMddHHmmss");
													String sendPort = StringUtils.substring(
															body.getSendPort(),
															5); // 剔除端口号(固定前五位),保留扩展码
													JSONObject map = new JSONObject();

													map.put(String.valueOf(AppCanstant.SmsEntity.Srv_node),
															body.getSrvNode());
													map.put(String.valueOf(AppCanstant.SmsEntity.ServiceCode),
															body.getServiceCode());
													map.put(String.valueOf(AppCanstant.SmsEntity.SmType),
															"1");
													map.put(String.valueOf(AppCanstant.SmsEntity.User),
															body.getUserNumber());
													map.put(String.valueOf(AppCanstant.SmsEntity.Smcont),
															body.getMsgContent());
													map.put(String.valueOf(AppCanstant.SmsEntity.Reportflag),
															StringUtils.toString(body.getReportFlag()));
													map.put(String.valueOf(AppCanstant.SmsEntity.recvTime),
															recvTime);
													map.put(String.valueOf(AppCanstant.SmsEntity.RecvModuleName),
															smspfmModuleName);
													map.put(String.valueOf(AppCanstant.SmsEntity.SendPort),
															sendPort);
													map.put(String.valueOf(AppCanstant.SmsEntity.SerialSegment),
															head.getSerialNoPart1()
																	+ ","
																	+ head.getSerialNoPart2()
																	+ ","
																	+ head.getSerialNoPart3());
													map.put(String.valueOf(AppCanstant.SmsEntity.OrgSerialNo),
															body.getOrgSerialNo());
													map.put(String.valueOf(AppCanstant.SmsEntity.OrgSrvNode),
															body.getOrgSrvNode());
													map.put(String.valueOf(AppCanstant.SmsEntity.SendFeedbackResult),
															"0");
													map.put("serialNo",serialNo);
													String userNumber = body
															.getUserNumber();

													String sequn = serialNo
															+ ":" + userNumber;
													map.put("userNumber",userNumber);
													map.put("sequn",sequn);
													map.put("dataType", "Submit2");
													logger.error(map.toJSONString());
												}
												
											}
										});
							}
						});
					} else {
						try {
							TimeUnit.MILLISECONDS.sleep(100);
						} catch (InterruptedException e) {
						}
					}
				}
			}
		}).start();
	}

	private void saveSbt1() {

		new Thread(new Runnable() {

			@Override
			public void run() {

				while ((executorService != null && !executorService
						.isShutdown()) || sbt1Msg.getSize() > 0) {

					List<MessageSbt> list = sbt1Msg.poll();

					if (list != null && !list.isEmpty()) {

						executorSaveService.submit(new Runnable() {

							@Override
							public void run() {

								List<String> serialnos = new ArrayList<>();

								AppStaticConfigs.pipelineOpts
										.save(new ABExcuteCommands() {

											@Override
											public void excute(
													RedisConnection connection) {

												List<byte[]> seqNums = new ArrayList<byte[]>();

												for (int i = 0; i < list.size(); i++) {

													MessageSbt msg = list
															.get(i);

													Head head = (Head) msg
															.getHead();
													Submit body = (Submit) msg
															.getBody();

													String recvTime = DateTimeUtils.dateToStr(
															msg.getRecvTime(),
															"yyyyMMddHHmmss");
													String serialNo = head
															.getSerialNo();

													Set<String> userNumber = body
															.getUserNumber();
													for (String un : userNumber) {

														Map<byte[], byte[]> map = new HashMap<>();

														map.put(AppCanstant.SmsEntity.Srv_node,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(StringUtils
																				.toString(body
																						.getSrvNode())));
														map.put(AppCanstant.SmsEntity.ServiceCode,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(StringUtils
																				.toString(body
																						.getServiceCode())));
														map.put(AppCanstant.SmsEntity.SmType,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(StringUtils
																				.toString(1)));
														map.put(AppCanstant.SmsEntity.User,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(StringUtils
																				.toString(un)));
														map.put(AppCanstant.SmsEntity.Smcont,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(StringUtils
																				.toString(body
																						.getMsgContent())));
														map.put(AppCanstant.SmsEntity.Reportflag,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(StringUtils
																				.toString(body
																						.getReportFlag())));
														map.put(AppCanstant.SmsEntity.recvTime,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(StringUtils
																				.toString(recvTime)));
														map.put(AppCanstant.SmsEntity.RecvModuleName,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(StringUtils
																				.toString(smspfmModuleName)));
														map.put(AppCanstant.SmsEntity.SendPort,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(StringUtils
																				.toString(body
																						.getSendPort())));
														map.put(AppCanstant.SmsEntity.SerialSegment,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(StringUtils
																				.toString(head
																						.getSerialNoPart1()
																						+ ","
																						+ head.getSerialNoPart2()
																						+ ","
																						+ head.getSerialNoPart3())));
														map.put(AppCanstant.SmsEntity.respresult,
																AppStaticConfigs.stringRedisSerializer
																		.serialize("0"));
														map.put(AppCanstant.SmsEntity.resptime,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(DateTimeUtils
																				.dateToStr(
																						new Date(),
																						"yyyyMMddHHmmss")));
														map.put(AppCanstant.SmsEntity.OrgSerialNo,
																AppStaticConfigs.stringRedisSerializer
																		.serialize(serialNo));
														String sequn = serialNo
																+ ":" + un;

														connection
																.hMSet(AppStaticConfigs.stringRedisSerializer
																		.serialize(iSMCacheHashPrefix
																				+ ":"
																				+ sequn),
																		map);
														// connection.lPush(iSMCacheB,
														// AppStaticConfigs.stringRedisSerializer.serialize(sequn));
														seqNums.add(AppStaticConfigs.stringRedisSerializer
																.serialize(sequn));
													}

													// connection.zAdd(AppStaticConfigs.iRptRecvSerialsZSetB,
													// System.currentTimeMillis(),
													// AppStaticConfigs.stringRedisSerializer.serialize(msg.getEncodeSerialNo()));

													serialnos.add(serialNo);
												}

												connection
														.lPush(iSMCacheB,
																seqNums.toArray(new byte[1][]));
											}

											@Override
											public void excuteCallback(
													List<Object> results) {
												logger.acceptd(serialnos.size()
														+ " saved "
														+ serialnos.toString());
											}
											
											@Override
											public void excuteFailCallback() {
												for (int i = 0; i < list.size(); i++) {
													MessageSbt msg = list.get(i);
													Head head = (Head) msg
															.getHead();
													Submit body = (Submit) msg
															.getBody();
													String serialNo = head.getSerialNo();
													if(serialnos.contains(serialNo)){
														continue;
													}
													String recvTime = DateTimeUtils.dateToStr(
															msg.getRecvTime(),
															"yyyyMMddHHmmss");
													String sendPort = StringUtils.substring(
															body.getSendPort(),
															5); // 剔除端口号(固定前五位),保留扩展码
													JSONObject map = new JSONObject();

													map.put(String.valueOf(AppCanstant.SmsEntity.Srv_node),
															body.getSrvNode());
													map.put(String.valueOf(AppCanstant.SmsEntity.ServiceCode),
															body.getServiceCode());
													map.put(String.valueOf(AppCanstant.SmsEntity.SmType),
															"1");
													JSONArray array = new JSONArray();
													for(String su: body.getUserNumber()){
														array.add(su);
													}
													map.put(String.valueOf(AppCanstant.SmsEntity.User),
															array);
													map.put(String.valueOf(AppCanstant.SmsEntity.Smcont),
															body.getMsgContent());
													map.put(String.valueOf(AppCanstant.SmsEntity.Reportflag),
															StringUtils.toString(body.getReportFlag()));
													map.put(String.valueOf(AppCanstant.SmsEntity.recvTime),
															recvTime);
													map.put(String.valueOf(AppCanstant.SmsEntity.RecvModuleName),
															smspfmModuleName);
													map.put(String.valueOf(AppCanstant.SmsEntity.SendPort),
															sendPort);
													map.put(String.valueOf(AppCanstant.SmsEntity.SerialSegment),
															head.getSerialNoPart1()
																	+ ","
																	+ head.getSerialNoPart2()
																	+ ","
																	+ head.getSerialNoPart3());
													
													map.put(String.valueOf(AppCanstant.SmsEntity.OrgSerialNo),
															serialNo);
													map.put(String.valueOf(AppCanstant.SmsEntity.SendFeedbackResult),
															"0");
													map.put("serialNo",serialNo);

													map.put("dataType", "Submit");
				
													logger.error(map.toJSONString());
												}
												
											}
											
										});
							}
						});
					} else {
						try {
							TimeUnit.MILLISECONDS.sleep(100);
						} catch (InterruptedException e) {
						}
					}
				}
			}
		}).start();
	}

	@Override
	public void handleSubmitMessage(Head head, Submit body, ExtSocket socket) {

		executorService.execute(new Runnable() {

			@Override
			public void run() {

				MessageSbt message = new MessageSbt(head, body);

				logger.debug(message.toString());

				String srvNode = body.getSrvNode();
				String serviceCode = body.getServiceCode();
				String sendPort = body.getSendPort();
				
				System.out.println(body.getMsgContent());

				if (isExtCodeUsable2(srvNode, sendPort)) { // 扩展码是否正确

					if (isSrvNodeService2(srvNode, serviceCode)) { // 业务端是否包含该业务类型

						if (AppStaticConfigs.speedControl) {
							// 短信被流量控制，结束流程
							synchronized (speedLock) {
								if (!smsSpeedControl(socket, message, srvNode)) {
									return;
								}
							}

						}

						// boolean isForbidTime = true;
						boolean isForbidTime = false; // 是否属于禁发时段

						if (AppStaticConfigs.forbidtimeFilterOn) {
							isForbidTime = isForbidTime2(srvNode, serviceCode);
						}

						if (!isForbidTime) {

							String encodeSerialNo = head.getSerialNo();

							boolean isRepeat = false; // 是否序号重复

							if (AppStaticConfigs.repeatseqsFilterOn) {
								isRepeat = filterQueue.contains(encodeSerialNo);
							}

							if (!isRepeat) {
								
								
								Submit sb = (Submit) message.getBody();
								int contentLength = sb.getMsgContent().length();
								if (contentLength > 0	&& contentLength <= AppStaticConfigs.smsContentMaxLength) {

									if (AppStaticConfigs.speedStatisticsOn) {
										increaseCurrSpeedS(1);
										increaseCurrSpeedM(1);
									}
									if (AppStaticConfigs.smsRuningFlag) {
										message.setRecvTime(new Date());
										message.setEncodeSerialNo(encodeSerialNo);

										sbt1Msg.put(message);

										logger.acceptd(message.toString());

									} else {
										smsOldPlatformSendMsg(head, body);
									}

									// 记录红名单序列号
									recordRedUser(head, body);

									socket.response(message, ErrorCode.CODE0);
								} else {
									logger.rejectd("短信内容过长错误:" + message.toString());
									socket.response(message, ErrorCode.CODE_6);
								}

							} else {
								logger.rejectd("重复发送错:" + message.toString());
								socket.response(message, ErrorCode.CODE_2);
							}
						} else {
							logger.rejectd("禁发时段错:" + message.toString());
							socket.response(message, ErrorCode.CODE_1);
						}
					} else {
						logger.rejectd("业务类型不匹配:" + message.toString());
						socket.response(message, ErrorCode.CODE_7);
					}
				} else {
					logger.rejectd("扩展码不匹配:" + message.toString());
					socket.response(message, ErrorCode.CODE_8);
				}

			}
		});
	}

	@Override
	public void handleSubmit2Message(Head head, Submit2 body, ExtSocket socket) {

		executorService.execute(new Runnable() {

			@Override
			public void run() {

				MessageSbt message = new MessageSbt(head, body);

				logger.debug(message.toString());

				String srvNode = body.getSrvNode();
				String serviceCode = body.getServiceCode();
				// String sendPort = StringUtils.substring(body.getSendPort(),
				// 5);

				if (isSrvNodeService2(srvNode, serviceCode)) {

					boolean isForbidTime = false; // 是否属于禁发时段

					if (AppStaticConfigs.forbidtimeFilterOn) {
						isForbidTime = isForbidTime2(srvNode, serviceCode);
					}

					if (!isForbidTime) {

						String encodeSerialNo = MD5Util.encode(body
								.getOrgSerialNo());

						boolean isRepeat = false; // 是否序号重复

						if (AppStaticConfigs.repeatseqsFilterOn) {
							isRepeat = filterQueue.contains(encodeSerialNo);
						}

						if (!isRepeat) {
							
							Submit2 sb = (Submit2) message.getBody();
							int contentLength = sb.getMsgContent().length();
							if (contentLength > 0
									&& contentLength <= AppStaticConfigs.smsContentMaxLength) {

								if (AppStaticConfigs.speedStatisticsOn) {
									increaseCurrSpeedS(1);
									increaseCurrSpeedM(1);
								}

								if (AppStaticConfigs.smsRuningFlag) {
									message.setRecvTime(new Date());
									message.setEncodeSerialNo(encodeSerialNo);

									sbt2Msg.put(message);

									logger.acceptd(message.toString());

								} else {
									smsOldPlatformSendMsg(head, body);
								}
								recordRedUser(head, body);
								socket.response(message, ErrorCode.CODE0);
							}else {
								logger.rejectd("短信内容长度错:" + message.toString());
								socket.response(message, ErrorCode.CODE_6);
							}

						} else {
							logger.rejectd("重复发送错:" + message.toString());
							socket.response(message, ErrorCode.CODE_2);
						}
					} else {
						logger.rejectd("禁发时段错:" + message.toString());
						socket.response(message, ErrorCode.CODE_1);
					}
				} else {
					logger.rejectd("业务类型不匹配:" + message.toString());
					socket.response(message, ErrorCode.CODE_7);
				}
			}
		});
	}

	/**
	 * 是否在禁发时间段内
	 * 
	 * @param srvNode
	 * @param serviceCode
	 * @return
	 */
	private boolean isForbidTime2(String srvNode, String serviceCode) {

		boolean result = false;

		Set<String> set = new HashSet<>(); // 所有适用的规则

		String ssrvNode = StringUtils.toString(srvNode);
		String sserviceCode = StringUtils.toString(serviceCode);

		Set<String> set2 = AppStaticConfigs.forbidTimeMap.get(ssrvNode + ":"
				+ sserviceCode);
		if (set2 != null) {
			set.addAll(set2); // 针对指定业务端的指定业务类型
		}
		Set<String> set3 = AppStaticConfigs.forbidTimeMap.get(ssrvNode + ":");
		if (set3 != null) {
			set.addAll(set3);// 针对指定业务端的所有业务
		}
		Set<String> set4 = AppStaticConfigs.forbidTimeMap.get(":"
				+ sserviceCode);
		if (set4 != null) {
			set.addAll(set4); // 针对制定业务类型的所有客户端
		}
		Set<String> set5 = AppStaticConfigs.forbidTimeMap.get(":");
		if (set5 != null) {
			set.addAll(set5); // 针对所有业务端的所有类型
		}

		List<String> forbidTime = null;
		for (String str : set) {
			if (StringUtils.isEmpty(str))
				continue;
			String[] strings = str.split(":");
			float startTime = StringUtils.strToFlaot(StringUtils
					.toString(strings[0]));
			float endTime = StringUtils.strToFlaot(StringUtils
					.toString(strings[1]));
			if (DateTimeUtils.isInInterval(startTime, endTime)) {
				// 序列号：手机号码: start-end,start-end
				if (forbidTime == null) {
					forbidTime = new ArrayList<>();
				}
				forbidTime.add(startTime + "-" + endTime);
			}
		}

		if (forbidTime != null) {
			result = true;
		}

		return result;
	}

	/**
	 * 判断SrvNode中是否包含业务类型ServiceCode
	 * 
	 * @param srvNode
	 * @param serviceCode
	 * @return
	 */
	private boolean isSrvNodeService2(String srvNode, String serviceCode) {
		boolean result = false;

		Set<String> serviceCodes = AppStaticConfigs.srvNSCodeMap.get(srvNode);

		if (serviceCodes != null && serviceCodes.contains(serviceCode)) {
			result = true;
		}

		return result;
	}

	/**
	 * 判断扩展码是否可用
	 * 
	 * @param srvNode
	 * @param sendPort
	 * @return
	 */
	private boolean isExtCodeUsable2(String srvNode, String sendPort) {

		boolean result = false;

		Set<String> extCodes = AppStaticConfigs.srvNodeExtCodeMap.get(srvNode);

		if (extCodes != null) {
			for (String ec : extCodes) {
				if (StringUtils.startsWith(sendPort, ec)
						|| StringUtils.isEmpty(ec)) {
					result = true;
					break;
				}
			}
		}

		return result;
	}

	private void startUpdateSpeedTasks() {

		new Thread(new Runnable() {

			@Override
			public void run() {

				while (executorService != null && !executorService.isShutdown()) {

					long speed = getAndResetCurrSpeedS();

					speeds.offer(Collections.singletonMap(CURR_ACCEPT_SPEEDS_B,
							AppStaticConfigs.stringRedisSerializer
									.serialize(speed + ":"
											+ DateTimeUtils.dateToStr2(null))));

					logger.debug("接口接收速度(条/s):" + speed);

					try {
						TimeUnit.MILLISECONDS.sleep(1000);
					} catch (InterruptedException e) {
					}
				}
			}
		}).start();

		new Thread(new Runnable() {

			@Override
			public void run() {

				int count = 0;

				while (executorService != null && !executorService.isShutdown()) {

					if (count >= 60) {

						long speed = getAndResetCurrSpeedM();

						speeds.offer(Collections.singletonMap(
								CURR_ACCEPT_SPEEDM_B,
								AppStaticConfigs.stringRedisSerializer
										.serialize(speed
												+ ":"
												+ DateTimeUtils
														.dateToStr2(null))));

						logger.debug("接口接收速度(条/m):" + speed);

						count = 0;

					} else {
						count++;
					}

					try {
						TimeUnit.MILLISECONDS.sleep(1000);
					} catch (InterruptedException e) {
					}
				}
			}
		}).start();
	}

	private void startSpeedWriteTasks() {

		new Thread(new Runnable() {

			@Override
			public void run() {

				while (executorService != null && !executorService.isShutdown()) {
					boolean isSleep = true;
					Map<byte[], byte[]> map = speeds.poll();
					if (map != null && !map.isEmpty()) {
						AppStaticConfigs.pipelineOpts
								.save(new ABExcuteCommands() {

									@Override
									public void excute(
											RedisConnection connection) {
										connection
												.hMSet(AppStaticConfigs.RunningParamsHashB,
														map);
										// 每秒的速率队列
										byte[] bytes = map
												.get(CURR_ACCEPT_SPEEDS_B);
										if (bytes != null) {
											connection.lPush(
													currAcceptSpeedS_List,
													bytes);
											connection
													.lTrim(currAcceptSpeedS_List,
															0,
															AppStaticConfigs.speedSLength - 1);
										}
										// 每分的速率队列
										byte[] bytem = map
												.get(CURR_ACCEPT_SPEEDM_B);
										if (bytem != null) {
											connection.lPush(
													currAcceptSpeedM_List,
													bytem);
											connection
													.lTrim(currAcceptSpeedM_List,
															0,
															AppStaticConfigs.speedMLength - 1);
										}
									}

									@Override
									public void excuteFailCallback() {
										
									}
								});
						isSleep = false;
					}
					if (isSleep) {
						try {
							TimeUnit.MILLISECONDS.sleep(1000);
						} catch (InterruptedException e) {
						}
					}
				}
			}
		}).start();
	}

	@Override
	public void destroy() {

		super.destroy();

		if (executorService != null && !executorService.isShutdown()) {
			executorService.shutdown();
		}

		while (!executorService.isShutdown()) {
			try {
				Thread.sleep(10);
			} catch (InterruptedException e) {
			}
		}

		if (executorService != null && executorService.isShutdown()) {
			executorService = null;
		}

		while (sbt1Msg.getSize() > 0 || sbt2Msg.getSize() > 0) {
			try {
				Thread.sleep(10);
			} catch (InterruptedException e) {
			}
		}

		if (executorSaveService != null && !executorSaveService.isShutdown()) {
			executorSaveService.shutdownNow();
		}

		if (sbt1Msg != null) {
			sbt1Msg.destroy();
		}

		if (sbt2Msg != null) {
			sbt2Msg.destroy();
		}

		if (redUserTread != null) {
			redUserTread.destroy();
			redUserTread = null;
		}
	}

	private long getAndResetCurrSpeedS() {
		long result = 0;
		synchronized (slock) {
			result = currSpeedS;
			currSpeedS = 0;
		}
		return result;
	}

	private void increaseCurrSpeedS(long count) {
		synchronized (slock) {
			this.currSpeedS += count;
		}
	}

	public long getCurrSpeedM() {
		return currSpeedM;
	}

	private long getAndResetCurrSpeedM() {
		long result = 0;
		synchronized (mlock) {
			result = currSpeedM;
			currSpeedM = 0;
		}
		return result;
	}

	private void increaseCurrSpeedM(long count) {
		synchronized (mlock) {
			this.currSpeedM += count;
		}
	}

	@Override
	public void handleUnBindMessage(Head head, UnBind body, ExtSocket socket) {
		if (socket != null && !socket.isClosed()) {
			logger.info("接收到UnBind消息:" + socket.getAccountInfo().toString());
			socket.response(new Message(head, body), ErrorCode.CODE0);
			try {
				TimeUnit.MILLISECONDS.sleep(1000);
			} catch (InterruptedException e1) {
			} finally {
				try {
					socket.close();
				} catch (Exception e) {
					logger.error("UnBind消息处理方法,关闭socket时出错了:" + e.getMessage());
				}
			}
		}
	}

	@Override
	public void handleActiveTestMessage(Head head, ActiveTest body,
			ExtSocket socket) {
		if (socket != null && !socket.isClosed()) {
			logger.info("接收到ActiveTest消息:" + socket.getAccountInfo().toString());
			socket.response(new Message(head, body), ErrorCode.CODE0);
		}
	}

	protected void smsOldPlatformSendMsg(Head head, Submit2 body) {
		MtOldSms mtOldSms = new MtOldSms();
		mtOldSms.setClientName(body.getSrvNode());
		mtOldSms.setSerialNo(head.getSerialNo());
		mtOldSms.setContent(body.getMsgContent());
		mtOldSms.setMesgType(body.getMsgType());
		mtOldSms.setMobileNo(body.getUserNumber());
		mtOldSms.setSendPort(body.getSendPort());
		mtOldSms.setSrr(1);
		mtOldSms.setWapUrl("");
		List<String> msgList = mtOldSms.toFileStringList();
		if (msgList != null) {
			for (int j = 0; j < msgList.size(); j++) {
				FileWriterTask.queue.offer(new FileRoute(
						FileRoute.KEY_SMS_OLD_SEND, msgList.get(j)));
			}
		}

	}

	private void smsOldPlatformSendMsg(Head head, Submit body) {
		// TODO 切换至旧平台处理模块
		// int conut = body.getUserCount();
		// for (int i = 0; i < conut; i++) {
		int i = 0;
		Set<String> set = body.getUserNumber();
		for (String userNumber : set) {
			MtOldSms mtOldSms = new MtOldSms();
			mtOldSms.setClientName(body.getSrvNode());
			mtOldSms.setSerialNo(head.getSerialNo());
			mtOldSms.setContent(body.getMsgContent());
			mtOldSms.setMesgType(body.getMsgType());
			mtOldSms.setMobileNo(userNumber);
			mtOldSms.setSendPort(body.getSendPort());
			mtOldSms.setSrr(1);
			mtOldSms.setWapUrl("");

			// 写入文件中
			// try {
			// AppStaticConfigs.fileManage.getOldSmsWriter().write(mtOldSms);
			List<String> msgList = mtOldSms.toFileStringList();
			if (msgList != null) {
				for (int j = 0; j < msgList.size(); j++) {
					FileWriterTask.queue.offer(new FileRoute(
							FileRoute.KEY_SMS_OLD_SEND, msgList.get(j)));
				}
			}
			// } catch (IOException e) {
			// logger.error("writer file error !!");
			// }
			// if(++i >= conut){
			// break;
			// }
		}
		// }
	}

	private boolean smsSpeedControl(ExtSocket socket, MessageSbt message,
			String srvNode) {
		// 流量控制功能开启
		boolean isFilter = true;
		Set<String> set = new HashSet<String>();
		Set<String> set1 = AppStaticConfigs.speedControlMap.get(srvNode);
		if (set1 != null) {
			set.addAll(set1);
		}

		Set<String> set2 = AppStaticConfigs.speedControlMap.get("*");
		if (set2 != null) {
			set.addAll(set2);
		}

		if (set.size() >= 0) {

			// 获取缓存流量数据
			final String key = srvNode;
			SmsSpeedStatistics smsSpeed = null;

			smsSpeed = AppStaticConfigs.smsSpeedMap.get(key);
			if (smsSpeed == null) {
				// 不在本地，去REDIS缓存中寻找
				List<SmsSpeedStatistics> smsList = new ArrayList<SmsSpeedStatistics>();
				AppStaticConfigs.pipelineOpts.save(new ABExcuteCommands() {
					@Override
					public void excute(RedisConnection connection) {
						String result = AppStaticConfigs.hashOpts.get(
								AppStaticConfigs.smsSpeedCache, key);
						if (!StringUtils.isEmpty(result)) {
							smsList.add(SmsSpeedStatistics.parseJson(result));
						}

					}

					@Override
					public void excuteFailCallback() {
						// TODO Auto-generated method stub
						
					}
				});

				// 缓存中没有，新建记录
				if (smsList.size() == 0) {
					System.out.println("未找到缓存数据： " + key);
					smsSpeed = new SmsSpeedStatistics();
				} else {
					smsSpeed = smsList.get(0);
				}
			}

			// 过滤

			for (String controls : set) {
				String[] controlBt = controls.split(":");
				if (controlBt.length >= 3) {
					// String limittype = controlBt[0];
					int flowspeed = StringUtils.toInteger(controlBt[1]);
					String timeunit = controlBt[2];
					long cnt = 0;
					Calendar cal = Calendar.getInstance();
					if (StringUtils.equals(timeunit, "1")) {// 秒
						cnt = smsSpeed.getSecondCnt(cal.get(Calendar.SECOND));
					} else if (StringUtils.equals(timeunit, "2")) {// 分
						cnt = smsSpeed.getMinuteCnt(cal.get(Calendar.MINUTE));
					} else if (StringUtils.equals(timeunit, "3")) {// 时
						cnt = smsSpeed
								.getHourCnt(cal.get(Calendar.HOUR_OF_DAY));
					} else if (StringUtils.equals(timeunit, "4")) {// 日
						cnt = smsSpeed.getDayCnt(cal.get(Calendar.DATE));
					} else if (StringUtils.equals(timeunit, "5")) {// 月
						cnt = smsSpeed.getMonthCnt(cal.get(Calendar.MONTH) + 1);
					}

					if (flowspeed <= cnt) {
						isFilter = false;
						break;
					}
				}
			}

			if (isFilter) {
				// 在规则之外则加入本地缓存流量记录
				AppStaticConfigs.smsSpeedMap.put(key, smsSpeed.add());

			} else {
				logger.rejectd("流量控制:" + message.toString());
				socket.response(message, ErrorCode.CODE7);
				return isFilter;
			}

		}
		return isFilter;
	}

	private void recordRedUser(Head head, Submit body) {
		Set<String> phones = new HashSet<String>();
		Set<String> set = body.getUserNumber();
		for (String key : set) {
			if (AppStaticConfigs.redUserMap.containsKey(key)) {
				phones.add(key);
			}
		}
		if (phones.size() > 0) {
			// 加入到缓存队列中完事
			for (String phone : phones) {
				redUserMsg.offer(head.getSerialNo() + ":" + phone);
			}
		}
	}

	private void recordRedUser(Head head, Submit2 body) {
		String phone = body.getUserNumber();

		if (AppStaticConfigs.redUserMap.containsKey(phone)) {
			redUserMsg.offer(head.getSerialNo() + ":" + phone);
		}

	}

	private void redUserSave() {
		List<byte[]> values = new ArrayList<byte[]>();
		while (redUserMsg.size() > 0) {
			String str = redUserMsg.poll();
			if (!StringUtils.isEmpty(str)) {
				values.add(AppStaticConfigs.stringRedisSerializer
						.serialize(str));
			} else {
				break;
			}
		}
		if (values.size() > 0) {

			AppStaticConfigs.pipelineOpts.save(new ABExcuteCommands() {

				@Override
				public void excute(RedisConnection connection) {
					connection.lPush(AppStaticConfigs.isRedUserSmsSerialB,
							values.toArray(new byte[1][]));
				}

				@Override
				public void excuteFailCallback() {
					// TODO Auto-generated method stub
					
				}

			});
		}
	}

	private class RedUserTask extends Thread {

		private long cnt = 0;

		@Override
		public void run() {
			while (executoRedUserrService != null
					&& !executoRedUserrService.isShutdown()
					&& !this.isInterrupted()) {
				
				cnt++;
				long time = System.currentTimeMillis();
				long timeCnt = time - redUserCurrTime;
				if (redUserMsg.size() >= redUserSize
						|| timeCnt > redUserMaxTime) {
					redUserCurrTime = time;
					executoRedUserrService.execute(new Runnable() {
						@Override
						public void run() {
							redUserSave();

						}
					});

				}
				if (AppStaticConfigs.smsRedUserReportTaskOnOff) {
					// 追踪报告　30分钟一次 //TODO
					if (cnt * redUserSleepTime > 1800 * 1000) {
						// if(cnt*redUserSleepTime >30*1000){
						cnt = 0;
						executoRedUserrService.execute(new Runnable() {

							@Override
							public void run() {
								boolean ruingFlag = false;
								do {
									ruingFlag = false;
									// 获取序列号
									RedisScript<List> script = new DefaultRedisScript<List>(
											AppStaticConfigs.luaGetRedUserList,
											List.class);

									List<String> keys = new ArrayList<>();
									keys.add(AppStaticConfigs.isRedUserSmsSerial);

									List<String> entities = AppStaticConfigs.scriptOpts
											.getList(script, keys, 1000 + "");
									List<byte[]> unfinishedList = new ArrayList<byte[]>();
									Calendar cal = Calendar.getInstance();
									cal.add(Calendar.HOUR, -24);// 最长查询24小时内
									Date queryDate = cal.getTime();
									for (int i = 0; i < entities.size(); i++) {
										String source = entities.get(i);
										List<String> strs = StringUtils
												.splitToList(source, ":");
										String phone = null;
										String serial = null;
										if (strs != null && strs.size() >= 2) {
											serial = strs.get(0);
											phone = strs.get(1);
										}
										if (phone == null || serial == null) {
											continue;
										}

										// 查询数据库
										SmsSendRecordExample smsRecordExample = new SmsSendRecordExample();
										smsRecordExample
												.createCriteria()
												.andRecvTimeGreaterThan(
														queryDate)
												.andSmSeqEqualTo(serial);
										List<SmsSendRecord> smsRcords = ContextUtils
												.getBean2(
														SmsSendRecordMapper.class)
												.selectByExample(
														smsRecordExample);

										if (smsRcords != null
												&& smsRcords.size() > 0) {
											SmsSendRecord rcord = smsRcords
													.get(0);
											String tempPhone = rcord
													.getUserNo();
											String svrNode = rcord.getSvrNode();
											if (StringUtils.isEmpty(svrNode)) {
												svrNode = "00000";
											}
											// TODO 写入特定文件中　
											String value = new RedUserRecord(
													rcord).toFileString();
											FileRoute fileRoute = new FileRoute(
													FileRoute.KEY_RED_USER_REPORT
															+ svrNode, value);
											FileWriterTask.queue
													.offer(fileRoute);
										} else {
											unfinishedList
													.add(AppStaticConfigs.stringRedisSerializer
															.serialize(source));
										}
									}
									if (unfinishedList.size() > 0) { // 未搜索到的写回到缓存队列中
										AppStaticConfigs.pipelineOpts
												.save(new ABExcuteCommands() {

													@Override
													public void excute(
															RedisConnection connection) {
														connection
																.lPush(AppStaticConfigs.isRedUserSmsSerialB,
																		unfinishedList
																				.toArray(new byte[1][]));
													}

													@Override
													public void excuteFailCallback() {
														// TODO Auto-generated
														// method stub

													}

												});
									}
									int size = entities.size();
									int size2 = unfinishedList.size();

									ruingFlag = size > 0 && size > size2;

								} while (ruingFlag);
							}
						});
					}
				}

				try {
					Thread.sleep(redUserSleepTime);
				} catch (InterruptedException e) {
				}

			}
		}

		@Override
		public void destroy() {
			while (redUserMsg.size() > 0) {
				try {
					Thread.sleep(50);
				} catch (InterruptedException e) {
				}
			}
			this.interrupt();
			if (executoRedUserrService != null
					&& !executoRedUserrService.isShutdown()) {
				executoRedUserrService.shutdown();
				executoRedUserrService = null;
			}

		}

	}

}
